from __future__ import annotations

import asyncio
import json
import logging
import datetime as dt
import re
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple, TypedDict
from zoneinfo import ZoneInfo

import httpx
from pydantic import BaseModel, Field
from sqlalchemy import select, update, func

from ..config import get_settings
from ..db import SessionLocal
from ..models import (
    AIAutoDecision,
    RSSFeedItem,
    RunMode,
    utcnow,
)
from ..routers.ai import ChatReq, chat
from ..routers.ai import _flatten_mcp_result  # 复用已有的 MCP 输出格式化
from .mcp_session import mcp
from .rss_ingestor import rss_ingestor


logger = logging.getLogger(__name__)
settings = get_settings()
try:
    LOCAL_TZ = ZoneInfo(settings.timezone)
except Exception:  # noqa: BLE001
    LOCAL_TZ = dt.timezone.utc

CANDLE_SPECS: List[Tuple[str, int]] = [
    ("10s", 6),
    ("1m", 6),
    ("15m", 6),
    ("1h", 6),
    ("1d", 6),
]


def _to_float_static(value: Any) -> Optional[float]:
    if value is None or value == "":
        return None
    try:
        return float(value)
    except Exception:
        return None


def _normalize_candles_static(raw: Any, interval: str) -> List[Dict[str, Any]]:
    """将 Gate K 线接口返回标准化为统一结构。"""
    if not raw:
        return []
    normalized: List[Dict[str, Any]] = []
    for item in raw:
        ts = open_ = high_ = low = close = volume = None
        if isinstance(item, dict):
            ts = _to_float_static(item.get("timestamp") or item.get("t"))
            open_ = _to_float_static(item.get("open") or item.get("o"))
            high_ = _to_float_static(item.get("high") or item.get("h"))
            low = _to_float_static(item.get("low") or item.get("l"))
            close = _to_float_static(item.get("close") or item.get("c"))
            volume = _to_float_static(item.get("volume") or item.get("v"))
        elif isinstance(item, list):
            ts = _to_float_static(item[0]) if len(item) > 0 else None
            volume = _to_float_static(item[1]) if len(item) > 1 else None
            close = _to_float_static(item[2]) if len(item) > 2 else None
            high_ = _to_float_static(item[3]) if len(item) > 3 else None
            low = _to_float_static(item[4]) if len(item) > 4 else None
            open_ = _to_float_static(item[5]) if len(item) > 5 else None
        normalized.append({
            "interval": interval,
            "timestamp": ts,
            "open": open_,
            "high": high_,
            "low": low,
            "close": close,
            "volume": volume,
        })
    normalized.sort(key=lambda x: x.get("timestamp") or 0)
    return normalized


def _extract_first_number(text: Optional[str]) -> Optional[float]:
    if not text:
        return None
    match = re.search(r"-?\d+(?:\.\d+)?", text)
    if not match:
        return None
    try:
        return float(match.group(0))
    except Exception:
        return None


def _try_parse_json(text: Optional[str]) -> Optional[Any]:
    if not text or not isinstance(text, str):
        return None
    try:
        return json.loads(text)
    except Exception:
        return None


async def fetch_gate_kline_bundle(symbol: str) -> Dict[str, List[Dict[str, Any]]]:
    """
    调 Gate USDT 永续 REST 接口，抓 10s/1m/15m/1h/1d 各 6 根 K 线并去重。
    返回按周期分组的字典，值为标准化 bar 列表。
    """
    contract = symbol.replace("-", "_").upper()
    out: Dict[str, List[Dict[str, Any]]] = {tf: [] for tf, _ in CANDLE_SPECS}

    async with httpx.AsyncClient(timeout=10.0) as client:
        for interval, limit in CANDLE_SPECS:
            params = {"contract": contract, "interval": interval, "limit": limit}
            url = "https://api.gateio.ws/api/v4/futures/usdt/candlesticks"
            try:
                resp = await client.get(url, params=params)
                resp.raise_for_status()
                data = resp.json()
                normalized = _normalize_candles_static(data, interval)
                dedup: Dict[float, Dict[str, Any]] = {}
                for item in normalized:
                    ts = item.get("timestamp")
                    if ts is None:
                        continue
                    dedup[ts] = item
                deduped_list = list(dedup.values())
                deduped_list.sort(key=lambda x: x.get("timestamp") or 0)
                out[interval] = deduped_list
            except Exception as exc:  # noqa: BLE001
                logger.warning("[AI-AutoTrader] 获取 Gate K 线失败 interval=%s: %s", interval, exc)

    return out


def summarize_kline_bundle(merged_klines: Dict[str, List[Dict[str, Any]]]) -> str:
    """
    将多周期 K 线压缩为短摘要，便于放入提示词，避免上下文过长。
    """
    if not merged_klines:
        return "获取 K 线数据失败"

    parts: List[str] = []
    for tf in ["10s", "1m", "15m", "1h", "1d"]:
        bars = merged_klines.get(tf) or []
        if not bars:
            continue
        last = bars[-1]
        o = _to_float_static(last.get("open"))
        c = _to_float_static(last.get("close"))
        h = _to_float_static(last.get("high"))
        l = _to_float_static(last.get("low"))
        if o is None or c is None:
            continue
        ret = (c - o) / o * 100 if o != 0 else 0.0
        rng = None
        if h is not None and l is not None and c:
            rng = (h - l) / c * 100
        text = f"{tf} 收盘较开盘变动 {ret:.2f}%"
        if rng is not None:
            text += f"，振幅约 {rng:.2f}%"
        parts.append(text)

    if not parts:
        return "获取 K 线数据失败"
    summary = "；".join(parts)
    return f"多周期 K 线概况：{summary}。"


async def fetch_realtime_snapshot(
    spot_connector: str,
    perp_connector: str,
    trading_pair: str,
    account_name: str,
) -> Dict[str, Any]:
    """
    业务层按顺序调用 MCP 工具，获取实时行情/仓位/余额。
    顺序固定：get_prices -> get_funding_rate -> get_positions -> get_portfolio_balances
    """
    snapshot: Dict[str, Any] = {
        "spot_price": None,
        "funding_rate": None,
        "positions": None,
        "balances": None,
        "raw": {},
    }

    # 1) get_prices
    try:
        prices_res = await mcp.call_tool("get_prices", {
            "connector_name": spot_connector,
            "trading_pairs": [trading_pair],
        })
        prices_text = _flatten_mcp_result(prices_res)
        snapshot["raw"]["prices"] = prices_text
        snapshot["spot_price"] = _extract_first_number(prices_text)
    except Exception as exc:  # noqa: BLE001
        logger.warning("[AI-AutoTrader] get_prices 失败: %s", exc)

    # 2) get_funding_rate
    try:
        funding_res = await mcp.call_tool("get_funding_rate", {
            "connector_name": perp_connector,
            "trading_pair": trading_pair,
            "force_refresh": True,
        })
        funding_text = _flatten_mcp_result(funding_res)
        snapshot["raw"]["funding_rate"] = funding_text
        snapshot["funding_rate"] = _extract_first_number(funding_text)
    except Exception as exc:  # noqa: BLE001
        logger.warning("[AI-AutoTrader] get_funding_rate 失败: %s", exc)

    # 3) get_positions
    try:
        positions_res = await mcp.call_tool("get_positions", {
            "connector_names": [perp_connector],
        })
        positions_text = _flatten_mcp_result(positions_res)
        snapshot["raw"]["positions"] = positions_text
        snapshot["positions"] = _try_parse_json(positions_text) or positions_text
    except Exception as exc:  # noqa: BLE001
        logger.warning("[AI-AutoTrader] get_positions 失败: %s", exc)

    # 4) get_portfolio_balances
    try:
        balances_res = await mcp.call_tool("get_portfolio_balances", {
            "accounts": [account_name],
            "connectors": [spot_connector, perp_connector],
        })
        balances_text = _flatten_mcp_result(balances_res)
        snapshot["raw"]["balances"] = balances_text
        snapshot["balances"] = _try_parse_json(balances_text) or balances_text
    except Exception as exc:  # noqa: BLE001
        logger.warning("[AI-AutoTrader] get_portfolio_balances 失败: %s", exc)

    return snapshot


def summarize_realtime_snapshot(snapshot: Dict[str, Any]) -> str:
    """将实时行情/仓位/余额压缩为短摘要。"""
    if not snapshot:
        return "实时数据获取失败"

    spot = snapshot.get("spot_price")
    funding = snapshot.get("funding_rate")
    positions = snapshot.get("positions") or []
    balances = snapshot.get("balances") or []

    pos_count = 0
    if isinstance(positions, list):
        pos_count = len(positions)
    elif isinstance(positions, dict):
        pos_count = len(positions)

    total_usdt = 0.0
    if isinstance(balances, list):
        for b in balances:
            if isinstance(b, dict) and (b.get("asset") or b.get("symbol")) in {"USDT", "usd", "USD"}:
                avail = b.get("available") or b.get("free") or b.get("balance")
                val = _to_float_static(avail)
                if val is not None:
                    total_usdt += val

    spot_str = f"{spot}" if spot is not None else "未知"
    funding_str = f"{funding}" if funding is not None else "未知"
    return (
        f"实时数据概况：现货最新价格约 {spot_str}，永续资金费率 {funding_str}；"
        f"当前持仓 {pos_count} 个，可用 USDT 余额约 {total_usdt:.2f}。"
    )


class FeaturePayload(TypedDict, total=False):
    price: Optional[float]
    ret_1h: Optional[float]
    ret_24h: Optional[float]
    vol_24h: Optional[float]
    rsi_14: Optional[float]
    macd: Optional[float]
    macd_signal: Optional[float]
    vol_change_24h: Optional[float]
    kline_summary: str
    realtime_summary: str
    kline_bundle: Dict[str, List[Dict[str, Any]]]
    snapshot: Dict[str, Any]


def _get_closes(bundle: Dict[str, List[Dict[str, Any]]], tf: str) -> List[float]:
    bars = bundle.get(tf) or []
    closes: List[float] = []
    for b in bars:
        c = _to_float_static(b.get("close"))
        if c is not None:
            closes.append(c)
    return closes


def _calc_ret(bundle: Dict[str, List[Dict[str, Any]]], tf: str) -> Optional[float]:
    closes = _get_closes(bundle, tf)
    if len(closes) < 2:
        return None
    first, last = closes[0], closes[-1]
    if first == 0:
        return None
    return (last - first) / first


def _calc_vol(bundle: Dict[str, List[Dict[str, Any]]], tf: str) -> Optional[float]:
    bars = bundle.get(tf) or []
    vols: List[float] = []
    for b in bars:
        v = _to_float_static(b.get("volume"))
        if v is not None:
            vols.append(v)
    return sum(vols) if vols else None


def _calc_vol_change(bundle: Dict[str, List[Dict[str, Any]]], tf: str) -> Optional[float]:
    bars = bundle.get(tf) or []
    vols = [_to_float_static(b.get("volume")) for b in bars if _to_float_static(b.get("volume")) is not None]
    if len(vols) < 2 or vols[0] == 0:
        return None
    return (vols[-1] - vols[0]) / vols[0]


def _ema(values: List[float], period: int) -> Optional[List[float]]:
    if len(values) < period or period <= 0:
        return None
    k = 2 / (period + 1)
    ema_vals: List[float] = []
    ema_prev = sum(values[:period]) / period
    ema_vals.append(ema_prev)
    for v in values[period:]:
        ema_prev = v * k + ema_prev * (1 - k)
        ema_vals.append(ema_prev)
    return ema_vals


def _calc_macd(bundle: Dict[str, List[Dict[str, Any]]], tf: str) -> Tuple[Optional[float], Optional[float]]:
    closes = _get_closes(bundle, tf)
    if len(closes) < 26:
        return None, None
    ema12 = _ema(closes, 12)
    ema26 = _ema(closes, 26)
    if not ema12 or not ema26:
        return None, None
    # 对齐长度
    diff_len = min(len(ema12), len(ema26))
    macd_line_list = [ema12[-diff_len + i] - ema26[-diff_len + i] for i in range(diff_len)]
    if len(macd_line_list) < 9:
        return macd_line_list[-1], None
    signal_list = _ema(macd_line_list, 9)
    return macd_line_list[-1], signal_list[-1] if signal_list else None


def _calc_rsi(bundle: Dict[str, List[Dict[str, Any]]], tf: str, period: int = 14) -> Optional[float]:
    closes = _get_closes(bundle, tf)
    if len(closes) <= period:
        return None
    gains = []
    losses = []
    for i in range(1, len(closes)):
        delta = closes[i] - closes[i - 1]
        if delta >= 0:
            gains.append(delta)
            losses.append(0.0)
        else:
            gains.append(0.0)
            losses.append(-delta)
    if len(gains) < period:
        return None
    avg_gain = sum(gains[-period:]) / period
    avg_loss = sum(losses[-period:]) / period
    if avg_loss == 0:
        return 100.0
    rs = avg_gain / avg_loss
    return 100 - (100 / (1 + rs))


async def fetch_features(
    spot_connector: str,
    perp_connector: str,
    trading_pair: str,
    account_name: str,
    base_tf: str = "1h",
) -> FeaturePayload:
    bundle = await fetch_gate_kline_bundle(trading_pair)
    kline_summary = summarize_kline_bundle(bundle)

    snapshot = await fetch_realtime_snapshot(
        spot_connector=spot_connector,
        perp_connector=perp_connector,
        trading_pair=trading_pair,
        account_name=account_name,
    )
    realtime_summary = summarize_realtime_snapshot(snapshot)

    price = _to_float_static(snapshot.get("spot_price"))
    ret_1h = _calc_ret(bundle, "1h")
    ret_24h = _calc_ret(bundle, "1d")
    vol_24h = _calc_vol(bundle, "1d")
    vol_change_24h = _calc_vol_change(bundle, "1d")
    rsi_14 = _calc_rsi(bundle, base_tf, period=14)
    macd, macd_signal = _calc_macd(bundle, base_tf)

    return {
        "price": price,
        "ret_1h": ret_1h,
        "ret_24h": ret_24h,
        "vol_24h": vol_24h,
        "rsi_14": rsi_14,
        "macd": macd,
        "macd_signal": macd_signal,
        "vol_change_24h": vol_change_24h,
        "kline_summary": kline_summary,
        "realtime_summary": realtime_summary,
        "kline_bundle": bundle,
        "snapshot": snapshot,
    }


class AIAutoTraderConfig(BaseModel):
    """AI 自动交易配置参数"""

    run_id: str = Field(..., description="运行 ID，用于标识和暂停任务")
    account_name: str = Field(..., description="账户名称")
    connector_name: str = Field(..., description="现货连接器名称，如 gate_io")
    trading_pair: str = Field(..., description="交易对，如 BTC-USDT")
    max_notional_usd: float = Field(..., gt=0, description="最大名义金额限制")
    max_slippage_bps: int = Field(..., ge=0, description="最大滑点（基点）")
    max_daily_loss_usd: float = Field(..., ge=0, description="当日最大亏损限制")
    mode: RunMode = Field(RunMode.shadow, description="运行模式：shadow|paper|live")
    interval_seconds: int = Field(120, ge=30, le=3600, description="执行间隔秒数，默认 120s")


@dataclass
class AIAutoTrader:
    config: AIAutoTraderConfig
    task: Optional[asyncio.Task] = None
    running: bool = False
    _wake_event: asyncio.Event = field(default_factory=asyncio.Event, init=False, repr=False)

    async def start(self) -> None:
        if self.running:
            logger.warning("AI AutoTrader 已在运行: %s", self.config.run_id)
            return
        self.running = True
        self._wake_event.clear()
        self.task = asyncio.create_task(self._loop())
        logger.info("AI AutoTrader 已启动: %s", self.config.run_id)

    async def stop(self) -> None:
        self.running = False
        self._wake_event.set()
        if self.task:
            self.task.cancel()
            try:
                await self.task
            except asyncio.CancelledError:
                pass
            self.task = None
        logger.info("AI AutoTrader 已停止: %s", self.config.run_id)

    async def _loop(self) -> None:
        while self.running:
            try:
                await self.run_cycle()
            except Exception as exc:
                logger.error("AI AutoTrader 运行错误 (%s): %s", self.config.run_id, exc, exc_info=True)
            try:
                await asyncio.wait_for(
                    self._wake_event.wait(),
                    timeout=self.config.interval_seconds,
                )
            except asyncio.TimeoutError:
                continue
            finally:
                self._wake_event.clear()

    async def run_cycle(self) -> None:
        spot_connector, perp_connector = self._resolve_connectors(self.config.connector_name)

        features = await fetch_features(
            spot_connector=spot_connector,
            perp_connector=perp_connector,
            trading_pair=self.config.trading_pair,
            account_name=self.config.account_name,
            base_tf="1h",
        )

        kline_bundle = features.get("kline_bundle") or {}
        snapshot = features.get("snapshot") or {}
        candles_data = self._flatten_kline_bundle(kline_bundle)
        candles_summary = features.get("kline_summary") or ""
        snapshot_summary = features.get("realtime_summary") or ""

        history_summary = await self._build_history_summary()
        news_summary, news_ids = await self._prepare_news_context()
        run_state = await self._get_run_state(latest_price=features.get("price"), timeframe="1h")

        ctx = self._build_context(
            candles_summary=candles_summary,
            history_summary=history_summary,
            news_summary=news_summary,
            snapshot_summary=snapshot_summary,
            features=features,
            run_state=run_state,
        )
        prompt = self._build_prompt(ctx)

        execute_trade = self.config.mode == RunMode.live
        allowed_categories: Optional[List[str]] = [
            "market_data",
            "portfolio",
            "position",
        ]

        blocked_tools: Optional[List[str]] = None
        if execute_trade:
            allowed_categories.append("trading")
        else:
            blocked_tools = ["place_order", "close_position", "cancel_order"]

        req = ChatReq(
            prompt=prompt,
            execute=execute_trade,
            allowed_tool_categories=allowed_categories,
            blocked_tools=blocked_tools,
            max_tool_rounds=8,
            temperature=0.1,
        )

        logger.info("[AI-AutoTrader] 开始执行: run_id=%s", self.config.run_id)
        resp = await chat(req)
        logger.info(
            "[AI-AutoTrader] 完成: run_id=%s, finish_reason=%s, text=%s",
            self.config.run_id,
            resp.finish_reason,
            resp.text,
        )
        if resp.tool_invocations:
            logger.info("[AI-AutoTrader] 工具调用: %s", json.dumps(resp.tool_invocations, ensure_ascii=False))

        parsed_signal = self._parse_final_json(resp.text or "")
        if parsed_signal:
            logger.info("[AI-AutoTrader] 解析信号: %s", parsed_signal)
        else:
            logger.warning("[AI-AutoTrader] 未能解析模型信号（缺少 FINAL_JSON 或 JSON 格式错误）")

        await self._persist_decision(
            prompt,
            resp,
            candles_data,
            history_summary,
            news_summary,
            news_ids,
            snapshot,
            snapshot_summary,
            features,
            parsed_signal,
        )
        await self._mark_news_delivered(news_ids)

    def notify_feed_update(self) -> None:
        if self.running:
            self._wake_event.set()

    async def _get_run_state(self, latest_price: Optional[float], timeframe: str) -> Dict[str, Any]:
        """获取运行状态：轮次计数与最近一次决策概要。"""
        out: Dict[str, Any] = {"round_idx": 1, "last_price": latest_price, "timeframe": timeframe}
        async with SessionLocal() as session:
            total = await session.scalar(
                select(func.count(AIAutoDecision.id)).where(AIAutoDecision.run_id == self.config.run_id)
            )
            out["round_idx"] = int(total or 0) + 1

            result = await session.execute(
                select(AIAutoDecision)
                .where(AIAutoDecision.run_id == self.config.run_id)
                .order_by(AIAutoDecision.ts.desc())
                .limit(1)
            )
            last = result.scalar_one_or_none()
            if last:
                out["last_time"] = last.ts.isoformat()
                parsed = (last.request_params or {}).get("parsed_signal") if last.request_params else None
                if not parsed and last.response_text:
                    parsed = self._parse_final_json(last.response_text)
                out["last_signal"] = (parsed or {}).get("signal", "未知")
                out["last_confidence"] = (parsed or {}).get("confidence", "未知")
        return out

    def _flatten_kline_bundle(self, bundle: Dict[str, List[Dict[str, Any]]]) -> List[Dict[str, Any]]:
        """将按周期分组的 K 线展开为列表，便于存储/截断。"""
        flat: List[Dict[str, Any]] = []
        for interval, bars in (bundle or {}).items():
            for bar in bars:
                # 确保 interval 字段存在
                if "interval" not in bar:
                    bar = {**bar, "interval": interval}
                flat.append(bar)
        flat.sort(key=lambda x: (x.get("interval") or "", x.get("timestamp") or 0))
        return flat

    def _build_context(
        self,
        candles_summary: str,
        history_summary: str,
        news_summary: str,
        snapshot_summary: str,
        features: FeaturePayload,
        run_state: Dict[str, Any],
    ) -> Dict[str, str]:
        spot_connector, perp_connector = self._resolve_connectors(self.config.connector_name)

        context = {
            "account_name": self.config.account_name,
            "spot_connector": spot_connector,
            "perp_connector": perp_connector,
            "trading_pair": self.config.trading_pair,
            "max_notional_usd": str(self.config.max_notional_usd),
            "max_slippage_bps": str(self.config.max_slippage_bps),
            "max_daily_loss_usd": str(self.config.max_daily_loss_usd),
            "mode": self.config.mode.value,
            "candles_summary": candles_summary,
            "history_text": history_summary,
            "news_summary": news_summary,
            "snapshot_summary": snapshot_summary,
            "features_text": self._format_features_for_prompt(features),
            "features": features,
            "agent_id": self.config.run_id,
            "history_summary": history_summary or "最近没有历史决策记录。",
            "last_time": run_state.get("last_time") or "未知",
            "last_signal": run_state.get("last_signal") or "未知",
            "last_confidence": run_state.get("last_confidence") or "未知",
            "last_price": run_state.get("last_price") or 0.0,
            "round_idx": run_state.get("round_idx") or 1,
            "symbol": self.config.trading_pair,
            "timeframe": run_state.get("timeframe") or "1h",
        }
        return context

    def _format_features_for_prompt(self, features: FeaturePayload) -> str:
        if not features:
            return "关键特征获取失败"
        parts: List[str] = []
        price = features.get("price")
        if price is not None:
            parts.append(f"现货价格≈{price}")
        if features.get("ret_1h") is not None:
            parts.append(f"1h 涨跌幅 {features['ret_1h']*100:.2f}%")
        if features.get("ret_24h") is not None:
            parts.append(f"24h 涨跌幅 {features['ret_24h']*100:.2f}%")
        if features.get("rsi_14") is not None:
            parts.append(f"RSI14={features['rsi_14']:.2f}")
        if features.get("macd") is not None:
            parts.append(f"MACD={features['macd']:.4f}")
        if features.get("macd_signal") is not None:
            parts.append(f"Signal={features['macd_signal']:.4f}")
        if features.get("vol_change_24h") is not None:
            parts.append(f"24h 成交量变化 {features['vol_change_24h']*100:.2f}%")
        if not parts:
            return "关键特征获取失败"
        return "；".join(parts)

    def _parse_final_json(self, output: str) -> Optional[Dict[str, Any]]:
        """解析模型输出中的 FINAL_JSON 段，返回 dict，失败返回 None。"""
        if "### FINAL_JSON" not in output:
            return None
        try:
            _, json_part = output.split("### FINAL_JSON", 1)
            json_str = json_part.strip()
            first = json_str.find("{")
            last = json_str.rfind("}")
            if first == -1 or last == -1 or last <= first:
                return None
            json_str = json_str[first : last + 1]
            data = json.loads(json_str)
            return data
        except Exception as exc:  # noqa: BLE001
            logger.warning("[AI-AutoTrader] FINAL_JSON 解析失败: %s", exc)
            return None

    def _build_prompt(self, context: Dict[str, Any]) -> str:
        def fmt_num(val: Optional[float]) -> str:
            return f"{val:.2f}" if isinstance(val, (int, float)) and val is not None else "未知"

        features: FeaturePayload = context.get("features") or {}
        return f"""
你是一个加密货币量化交易分析助手，负责针对单一交易对给出离散的交易信号（BUY / SELL / HOLD）。

当前 Agent 信息：
- Agent ID：{context.get('agent_id')}

历史决策统计（仅供参考）：
{context.get('history_summary')}

历史决策信息（上一轮）：
- 上一次决策时间：{context.get('last_time')}
- 上一次信号：{context.get('last_signal')}
- 上一次信心度：{context.get('last_confidence')}
- 上一次价格：{fmt_num(context.get('last_price'))} USDT

K 线摘要（由系统根据 Gate USDT 永续多周期 K 线预先计算）：
{context.get('candles_summary')}

实时账户与市场状态摘要（由系统预先按固定顺序依次调用 get_prices、get_funding_rate、get_positions、get_portfolio_balances 得到）：
{context.get('snapshot_summary')}

当前轮次：第 {context.get('round_idx')} 轮
当前数值特征（{context.get('symbol')}, {context.get('timeframe')} 周期）：
- 当前价格：{fmt_num(features.get('price'))} USDT
- 最近 1 小时收益率：{fmt_num((features.get('ret_1h') or 0)*100)}%
- 最近 24 小时收益率：{fmt_num((features.get('ret_24h') or 0)*100)}%
- 最近 24 小时收益率波动率（标准差）：{fmt_num(features.get('vol_24h'))}
- 14 周期 RSI：{fmt_num(features.get('rsi_14'))}
- MACD 值：{fmt_num(features.get('macd'))}
- MACD Signal 值：{fmt_num(features.get('macd_signal'))}
- 最近 24 小时成交量变化：{fmt_num((features.get('vol_change_24h') or 0)*100)}%

重要约束（请务必严格遵守）：
1. 先用中文给出你的思考过程，说明你如何综合“历史决策”“K 线摘要”“实时状态摘要”和上述数值特征做出判断。
2. 思考过程严格控制在 4~6 句，总字数不要超过 120 个汉字。
3. 思考过程结束后，单独开启一行，输出如下固定标记（不要加多余符号）：
   ### FINAL_JSON
4. 在该标记的下一行，只输出一个 JSON 对象，格式必须为（注意是合法 JSON）：

{{
  "signal": "BUY 或 SELL 或 HOLD（必须大写英文）",
  "confidence": 整数（0 到 100）,
  "reason": "简短中文理由"
}}

5. JSON 必须满足以下要求：
   - 最外层只能有一个花括号 {{ ... }}，不能有第二个 JSON。
   - key 必须用半角双引号："signal"、"confidence"、"reason"。
   - value 中不要使用中文引号或单引号，不要出现注释。
   - 每一行末尾不能多一个逗号，最后一行后面不能有逗号。
   - "reason" 的内容尽量控制在 40 个汉字以内，不能换行，不能再包含引号。

6. 在输出 JSON 之后，不要再输出任何文字或符号（不要再写解释、不要写 think、不要写 ```）。
7. 即使你在思考过程中已经给出了结论，也必须再用上述 JSON 格式完整输出一次结论。
"""

    async def _persist_decision(
        self,
        prompt: str,
        resp,
        candles_data: Optional[List[Dict[str, Any]]],
        history_text: str,
        news_summary: str = "",
        news_ids: Optional[List[int]] = None,
        snapshot: Optional[Dict[str, Any]] = None,
        snapshot_summary: str = "",
        features: Optional[FeaturePayload] = None,
        parsed_signal: Optional[Dict[str, Any]] = None,
    ) -> None:
        spot_connector, perp_connector = self._resolve_connectors(self.config.connector_name)
        record = AIAutoDecision(
            run_id=self.config.run_id,
            account_name=self.config.account_name,
            connector_name=spot_connector,
            trading_pair=self.config.trading_pair,
            mode=self.config.mode,
            prompt=prompt.strip(),
            request_params={
                "account_name": self.config.account_name,
                "connector_name": self.config.connector_name,
                "spot_connector": spot_connector,
                "perp_connector": perp_connector,
                "trading_pair": self.config.trading_pair,
                "max_notional_usd": self.config.max_notional_usd,
                "max_slippage_bps": self.config.max_slippage_bps,
                "max_daily_loss_usd": self.config.max_daily_loss_usd,
                "interval_seconds": self.config.interval_seconds,
                "mode": self.config.mode.value,
                "history_context": history_text,
                "candles_points": len(candles_data) if candles_data else 0,
                "candle_specs": [[interval, count] for interval, count in CANDLE_SPECS],
                "news_context": news_summary,
                "news_item_ids": news_ids or [],
                "snapshot_summary": snapshot_summary,
                "snapshot": snapshot or {},
                "features": features or {},
                "parsed_signal": parsed_signal or {},
            },
            response_text=resp.text,
            finish_reason=resp.finish_reason,
            tool_invocations=resp.tool_invocations or None,
            tokens_used=resp.tokens_used,
            candles=self._truncate_candles(candles_data),
        )

        try:
            async with SessionLocal() as session:
                session.add(record)
                await session.commit()
        except Exception as exc:
            logger.error("[AI-AutoTrader] 保存决策失败: %s", exc, exc_info=True)

    def _resolve_connectors(self, raw: str) -> Tuple[str, str]:
        name = (raw or "").strip().lower()
        if not name:
            raise ValueError("connector_name 不能为空")

        alias_map = {
            "gate": "gate_io",
            "gateio": "gate_io",
            "gate_io": "gate_io",
            "binance": "binance",
            "okx": "okx",
        }

        spot = alias_map.get(name, name)
        if spot.endswith("_perpetual"):
            spot = spot[: -len("_perpetual")]

        perp_alias_map = {
            "gate": "gate_io_perpetual",
            "gateio": "gate_io_perpetual",
            "gate_io": "gate_io_perpetual",
        }

        perp = perp_alias_map.get(name)
        if not perp:
            perp = f"{spot}_perpetual"

        return spot, perp

    async def _fetch_candles(self) -> Optional[List[Dict[str, Any]]]:
        contract = self.config.trading_pair.replace("-", "_").upper()
        all_records: List[Dict[str, Any]] = []

        async with httpx.AsyncClient(timeout=10.0) as client:
            for interval, limit in CANDLE_SPECS:
                params = {
                    "contract": contract,
                    "interval": interval,
                    "limit": limit,
                }
                url = "https://api.gateio.ws/api/v4/futures/usdt/candlesticks"
                try:
                    resp = await client.get(url, params=params)
                    resp.raise_for_status()
                    data = resp.json()
                    normalized = self._normalize_candles(data, interval)
                    all_records.extend(normalized)
                except Exception as exc:
                    logger.warning("[AI-AutoTrader] 获取 Gate K 线失败 interval=%s: %s", interval, exc)

        if not all_records:
            return None

        dedup: Dict[Tuple[str, Optional[float]], Dict[str, Any]] = {}
        for item in all_records:
            key = (item.get("interval"), item.get("timestamp"))
            dedup[key] = item

        merged = list(dedup.values())
        merged.sort(key=lambda x: (x.get("interval") or "", x.get("timestamp") or 0))
        return merged

    def _normalize_candles(self, raw: Any, interval: str) -> List[Dict[str, Any]]:
        if not raw:
            return []
        normalized: List[Dict[str, Any]] = []
        for item in raw:
            ts = open_ = high = low = close = volume = None
            if isinstance(item, dict):
                ts = self._to_float(item.get("timestamp") or item.get("t"))
                open_ = self._to_float(item.get("open") or item.get("o"))
                high = self._to_float(item.get("high") or item.get("h"))
                low = self._to_float(item.get("low") or item.get("l"))
                close = self._to_float(item.get("close") or item.get("c"))
                volume = self._to_float(item.get("volume") or item.get("v"))
            elif isinstance(item, list):
                ts = self._to_float(item[0]) if len(item) > 0 else None
                volume = self._to_float(item[1]) if len(item) > 1 else None
                close = self._to_float(item[2]) if len(item) > 2 else None
                high = self._to_float(item[3]) if len(item) > 3 else None
                low = self._to_float(item[4]) if len(item) > 4 else None
                open_ = self._to_float(item[5]) if len(item) > 5 else None
            normalized.append({
                "interval": interval,
                "timestamp": ts,
                "open": open_,
                "high": high,
                "low": low,
                "close": close,
                "volume": volume,
            })
        normalized.sort(key=lambda x: x.get("timestamp") or 0)
        return normalized

    def _summarize_candles(self, candles: Optional[List[Dict[str, Any]]]) -> str:
        if not candles:
            return "获取 K 线数据失败"
        lines: List[str] = []
        for interval, _ in CANDLE_SPECS:
            subset = [c for c in candles if c.get("interval") == interval][-6:]
            if not subset:
                continue
            lines.append(f"[{interval}] 最近 {len(subset)} 根")
            for item in subset:
                ts = item.get("timestamp")
                if ts:
                    try:
                        ts_str = dt.datetime.fromtimestamp(float(ts), tz=dt.timezone.utc).isoformat()
                    except Exception:
                        ts_str = str(ts)
                else:
                    ts_str = "N/A"
                lines.append(
                    f"  {ts_str} | open={item.get('open')} | close={item.get('close')} | high={item.get('high')} | low={item.get('low')} | vol={item.get('volume')}"
                )
        return "\n".join(lines)

    def _truncate_candles(self, candles: Optional[List[Dict[str, Any]]]) -> Optional[List[Dict[str, Any]]]:
        if not candles:
            return None
        # 保留每个 interval 最新 50 条，避免无限增长
        keep: List[Dict[str, Any]] = []
        counts: Dict[str, int] = {}
        for item in reversed(candles):  # 从最新开始
            interval = item.get("interval") or "unknown"
            counts.setdefault(interval, 0)
            if counts[interval] < 50:
                keep.append(item)
                counts[interval] += 1
        keep.reverse()
        keep.sort(key=lambda x: (x.get("interval") or "", x.get("timestamp") or 0))
        return keep

    def _to_float(self, value: Any) -> Optional[float]:
        if value is None or value == "":
            return None
        try:
            return float(value)
        except Exception:
            return None

    async def _build_history_summary(self, window: int = 5) -> str:
        """统计最近 window 次决策的 BUY/SELL/HOLD 次数。"""
        try:
            async with SessionLocal() as session:
                stmt = (
                    select(AIAutoDecision)
                    .where(AIAutoDecision.run_id == self.config.run_id)
                    .order_by(AIAutoDecision.ts.desc())
                    .limit(window)
                )
                result = await session.execute(stmt)
                records = list(result.scalars().all())
        except Exception as exc:
            logger.warning("[AI-AutoTrader] 获取历史决策失败: %s", exc)
            return "最近没有历史决策记录。"

        if not records:
            return "最近没有历史决策记录。"

        buys = sells = holds = 0
        for rec in records:
            signal = None
            try:
                signal = (rec.request_params or {}).get("parsed_signal", {}).get("signal")
            except Exception:
                signal = None
            if not signal and rec.response_text:
                parsed = self._parse_final_json(rec.response_text)
                if parsed:
                    signal = parsed.get("signal")
            signal_upper = (signal or "").strip().upper()
            if signal_upper == "BUY":
                buys += 1
            elif signal_upper == "SELL":
                sells += 1
            elif signal_upper == "HOLD":
                holds += 1

        total = len(records)
        return f"最近 {total} 次决策统计：BUY {buys} 次，SELL {sells} 次，HOLD {holds} 次。"

    async def _prepare_news_context(self) -> Tuple[str, List[int]]:
        if not settings.rsshub_enabled or not settings.rsshub_feed_list:
            return "", []

        limit = max(1, min(settings.rsshub_items_per_fetch, 10))
        async with SessionLocal() as session:
            stmt = (
                select(RSSFeedItem)
                .where(RSSFeedItem.delivered.is_(False))
                .order_by(
                    RSSFeedItem.published_at.desc(),
                    RSSFeedItem.fetched_at.desc(),
                )
                .limit(limit)
            )
            result = await session.execute(stmt)
            records = list(result.scalars().all())

        if not records:
            return "", []

        summary = self._summarize_news(records)
        ids = [item.id for item in records if item.id is not None]
        return summary, ids

    def _summarize_news(self, records: List[RSSFeedItem]) -> str:
        lines: List[str] = []
        ordered = sorted(
            records,
            key=lambda item: (item.published_at or item.fetched_at or dt.datetime.min.replace(tzinfo=dt.timezone.utc)),
        )
        for item in ordered:
            ts = item.published_at or item.fetched_at
            ts_str = self._format_timestamp(ts)
            title = (item.title or "无标题").strip()
            summary = (item.summary or "").strip()
            if summary and len(summary) > 200:
                summary = summary[:200].rstrip() + "..."
            link = item.link or ""

            lines.append(f"- [{ts_str}] {title}")
            if summary:
                lines.append(f"  {summary}")
            if link:
                lines.append(f"  来源: {link}")
        return "\n".join(lines)

    def _format_timestamp(self, value: Optional[dt.datetime]) -> str:
        if not value:
            return "未知时间"
        try:
            localized = value.astimezone(LOCAL_TZ)
        except Exception:
            localized = value
        return localized.strftime("%Y-%m-%d %H:%M")

    async def _mark_news_delivered(self, item_ids: List[int]) -> None:
        if not item_ids:
            return
        async with SessionLocal() as session:
            stmt = (
                update(RSSFeedItem)
                .where(RSSFeedItem.id.in_(item_ids))
                .values(delivered=True, delivered_at=utcnow())
            )
            await session.execute(stmt)
            await session.commit()


_ai_traders: Dict[str, AIAutoTrader] = {}


async def start_ai_trader(config: AIAutoTraderConfig) -> None:
    if config.run_id in _ai_traders:
        logger.warning("AI AutoTrader 已存在，将先停止再重新启动: %s", config.run_id)
        trader = _ai_traders[config.run_id]
        await trader.stop()

    trader = AIAutoTrader(config=config)
    _ai_traders[config.run_id] = trader
    await trader.start()


async def stop_ai_trader(run_id: str) -> None:
    trader = _ai_traders.get(run_id)
    if not trader:
        logger.warning("AI AutoTrader 不存在: %s", run_id)
        return
    await trader.stop()
    _ai_traders.pop(run_id, None)


def list_ai_traders() -> List[Dict[str, str]]:
    out = []
    for run_id, trader in _ai_traders.items():
        status = "running" if trader.running else "stopped"
        out.append({
            "run_id": run_id,
            "status": status,
            "account_name": trader.config.account_name,
            "connector_name": trader.config.connector_name,
            "trading_pair": trader.config.trading_pair,
            "mode": trader.config.mode.value,
            "interval_seconds": trader.config.interval_seconds,
            "candle_specs": [[interval, count] for interval, count in CANDLE_SPECS],
        })
    return out


async def _rss_new_items_handler(items: List[RSSFeedItem]) -> None:
    if not items:
        return
    logger.info("[AI-AutoTrader] 收到 %s 条 RSS 新内容，唤醒所有运行中的任务", len(items))
    for trader in _ai_traders.values():
        trader.notify_feed_update()


rss_ingestor.set_on_new_items(_rss_new_items_handler)
